#![allow(unused_variables)]
#![allow(unused_imports)]
#![allow(clippy::redundant_clone)]
#![allow(clippy::or_fun_call)]

use dotenv::dotenv;
use std::env::var;

use base64;
use std::{
    pin::Pin,
    task::{Context, Poll},
};

use std::collections::{HashMap, HashSet};
use std::convert::Infallible;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use std::{borrow::Cow, fmt};

use serde::{Deserialize, Serialize};
use serde_json::json;

use http::version::Version;
use hyper::{service::make_service_fn, Server};
use tonic::{Code, Request, Response, Status};
use tower::Service;
use warp::reply::json;
use warp::{http::StatusCode, Filter, Reply};

use futures::future::{self, Either, TryFutureExt};

use tonic::transport::channel::Channel;

use self::utils::*;

use coordinator::AvscMetaRead;
use coordinator::{AvroBulkCreate, JsonBulkCreate};
use coordinator::{AvroGate, SchemaRead};
use coordinator::{
    ExchangeTokenCreate, PasswordTokenCreate, RefreshTokenCreate, RefreshTokenParams,
    SignatureTokenReply, TrustTokenCreate,
};
use xdefault::DefaultReply;

use coordinator::coordinator_server::{Coordinator, CoordinatorServer};
use coordinator::push_client::PushClient;
use coordinator::push_server::{Push, PushServer};

use coordinator::meta_client::MetaClient;
use coordinator::meta_server::{Meta, MetaServer};
use coordinator::query_client::QueryClient;
use coordinator::query_server::{Query, QueryServer};
use coordinator::signature_client::SignatureClient;
use coordinator::signature_server::{Signature, SignatureServer};

#[macro_use]
extern crate log;

type Error = Box<dyn std::error::Error + Send + Sync + 'static>;

pub mod xdefault {
    tonic::include_proto!("proto.xdefault");
}
pub mod coordinator {
    tonic::include_proto!("proto.push");
    tonic::include_proto!("proto.coordinator");
    tonic::include_proto!("proto.signature");
    tonic::include_proto!("proto.meta");

    tonic::include_proto!("proto.query");
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    env_logger::init();
    dotenv::from_filename("meta-lib-static/env/.env").ok();

    let hosts = var("COORDINATOR_GATEWAY_HOSTS").unwrap_or("127.0.0.1".to_string());
    let port = var("COORDINATOR_GATEWAY_PORT")
        .unwrap()
        .parse::<u16>()
        .unwrap();

    let o_ip = resolve_socket_addr(&hosts);
    if o_ip.is_none() {
        println!("COORDINATOR_GATEWAY_HOSTS not match! skip");
        return Ok(());
    }
    let addr = SocketAddr::new(o_ip.unwrap(), port);
    println!("start server at {}", addr);

    let push_online_hosts =
        var("ACCESS_GATEWAY_PUSH_ONLINE_HOSTS").unwrap_or("127.0.0.1".to_string());
    let push_staging_hosts =
        var("ACCESS_GATEWAY_PUSH_STAGING_HOSTS").unwrap_or("127.0.0.1".to_string());
    let push_port = var("ACCESS_GATEWAY_PUSH_PORT")?;
    let push_hosts = format!("{},{}", push_online_hosts, push_staging_hosts);

    let signature_hosts = var("SIGNATURE_SERVER_HOSTS").unwrap_or("127.0.0.1".to_string());
    let signature_port = var("SIGNATURE_SERVER_PORT")?;
    let meta_hosts = var("META_SERVER_HOSTS").unwrap_or("127.0.0.1".to_string());
    let meta_port = var("META_SERVER_PORT")?;
    let query_hosts = var("QUERY_SERVER_HOSTS").unwrap_or("127.0.0.1".to_string());
    let query_port = var("QUERY_SERVER_PORT")?;

    let push_server_nodes = build_nodes(&push_hosts, &push_port);
    let signature_server_nodes = build_nodes(&signature_hosts, &signature_port);
    let meta_server_nodes = build_nodes(&meta_hosts, &meta_port);
    let query_server_nodes = build_nodes(&query_hosts, &query_port);

    let client = states::mk_client(
        &push_server_nodes,
        &signature_server_nodes,
        &meta_server_nodes,
        &query_server_nodes,
    )
    .await;
    let tonic = CoordinatorServer::new(MyCoordinator {
        client: client.clone(),
    });

    use hyper::server::conn::AddrStream;
    Server::bind(&addr)
        .serve(make_service_fn(move |socket: &AddrStream| {
            let mut tonic = tonic.clone();

            let remote_ip = socket.remote_addr().ip().to_string();
            let mut warp = warp::service(filters::routes(client.clone(), &remote_ip));

            future::ok::<_, Infallible>(tower::service_fn(
                move |req: hyper::Request<hyper::Body>| match req.version() {
                    Version::HTTP_11 | Version::HTTP_10 => Either::Left(
                        warp.call(req)
                            .map_ok(|res| res.map(EitherBody::Left))
                            .map_err(Error::from),
                    ),
                    Version::HTTP_2 => Either::Right(
                        tonic
                            .call(req)
                            .map_ok(|res| res.map(EitherBody::Right))
                            .map_err(Error::from),
                    ),
                    _ => unimplemented!(),
                },
            ))
        }))
        .await?;

    Ok(())
}

#[derive(Debug)]
pub struct MyCoordinator {
    client: states::Client,
}
#[tonic::async_trait]
impl Coordinator for MyCoordinator {}
#[tonic::async_trait]
impl Signature for MyCoordinator {
    async fn create_password_token(
        &self,
        create: Request<PasswordTokenCreate>,
    ) -> Result<Response<SignatureTokenReply>, Status> {
        handlers::grpc_create_password_token(
            self.client.clone(),
            "".to_string(),
            create.into_inner(),
        )
        .await
    }
    async fn create_refresh_token(
        &self,
        create: Request<RefreshTokenCreate>,
    ) -> Result<Response<SignatureTokenReply>, Status> {
        handlers::grpc_create_refresh_token(self.client.clone(), create.into_inner()).await
    }
    async fn create_exchange_token(
        &self,
        create: Request<ExchangeTokenCreate>,
    ) -> Result<Response<SignatureTokenReply>, Status> {
        unimplemented!()
    }
    async fn create_trust_token(
        &self,
        create: Request<TrustTokenCreate>,
    ) -> Result<Response<SignatureTokenReply>, Status> {
        unimplemented!()
    }
}

#[tonic::async_trait]
impl Push for MyCoordinator {
    async fn create_staging_avro_bulk(
        &self,
        create: Request<AvroBulkCreate>,
    ) -> Result<Response<DefaultReply>, Status> {
        handlers::grpc_create_staging_avro_bulk(self.client.clone(), create.into_inner()).await
    }
    async fn create_staging_json_bulk(
        &self,
        create: Request<JsonBulkCreate>,
    ) -> Result<Response<DefaultReply>, Status> {
        handlers::grpc_create_staging_json_bulk(self.client.clone(), create.into_inner()).await
    }
    async fn create_online_avro_bulk(
        &self,
        create: Request<AvroBulkCreate>,
    ) -> Result<Response<DefaultReply>, Status> {
        handlers::grpc_create_online_avro_bulk(self.client.clone(), create.into_inner()).await
    }
    async fn create_online_json_bulk(
        &self,
        create: Request<JsonBulkCreate>,
    ) -> Result<Response<DefaultReply>, Status> {
        let response =
            handlers::grpc_create_online_json_bulk(self.client.clone(), create.into_inner())
                .await?;
        /// Reads a raw schema from a subject in the schema registry.
        ///
        /// Returns a reply with a default empty reply if successful.
        ///
        /// # Errors
        ///
        /// Returns a `Status` if the `SchemaRead` is invalid or if the
        /// schema registry is unreachable or returns an error.
        Ok(response)
    }
}
#[tonic::async_trait]
impl Meta for MyCoordinator {
    /*************  ✨ Codeium Command ⭐  *************/
    /******  6d0c2d7f-576c-49f9-a8cc-582fc7d816fa  *******/
    async fn read_raw_schema(
        &self,
        read: Request<SchemaRead>,
    ) -> Result<Response<DefaultReply>, Status> {
        unimplemented!()
    }
    async fn read_avsc_schema(
        &self,
        gen: Request<SchemaRead>,
    ) -> Result<Response<DefaultReply>, Status> {
        unimplemented!()
    }
    async fn gate_json_avro(
        &self,
        gate: Request<AvroGate>,
    ) -> Result<Response<DefaultReply>, Status> {
        unimplemented!()
    }
    async fn gate_bin_avro(
        &self,
        gate: Request<AvroGate>,
    ) -> Result<Response<DefaultReply>, Status> {
        unimplemented!()
    }
}
#[tonic::async_trait]
impl Query for MyCoordinator {
    async fn read_avsc_meta(
        &self,
        read: Request<AvscMetaRead>,
    ) -> Result<Response<DefaultReply>, Status> {
        unimplemented!()
    }
}

mod filters {
    use super::handlers::*;
    use super::*;
    use warp::Filter;

    pub fn routes(
        client: states::Client,
        ip: &str,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        let cors = warp::cors()
            .allow_any_origin()
            .allow_credentials(true)
            .allow_headers(vec![
                "Content-Type",
                "Authorization",
                "User-Agent",
                "Sec-Fetch-Mode",
                "Referer",
                "Origin",
                "X-Real-IP",
                "X-Forwarded-For",
                "Access-Control-Request-Method",
                "Access-Control-Request-Headers",
            ])
            .allow_methods(vec!["GET", "POST", "DELETE", "OPTIONS"]);

        coordinator_routes()
            .or(push_routes(client.clone()))
            .or(signature_routes(client.clone(), ip))
            .or(meta_routes(client.clone(), ip))
            .or(query_routes(client.clone(), ip))
            .with(cors)
    }
    pub fn coordinator_routes(
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        warp::path("coordinator").and(warp::path("health").map(|| StatusCode::OK))
    }
    pub fn with_grpc(
        client: states::Client,
    ) -> impl Filter<Extract = (states::Client,), Error = std::convert::Infallible> + Clone {
        warp::any().map(move || client.clone())
    }
    pub fn with_ip(
        ip: String,
    ) -> impl Filter<Extract = (String,), Error = std::convert::Infallible> + Clone {
        (warp::header::<String>("X-Real-IP"))
            .or(warp::any().map(move || ip.clone()))
            .unify()
    }

    pub fn signature_routes(
        client: states::Client,
        ip: &str,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        warp::path("signature").and(
            (warp::path("ping").and(warp::get()).map(|| "pong"))
                .or(signature_password_token(client.clone(), ip.to_string()))
                .or(signature_refresh_token(client.clone(), ip.to_string()))
                .or(signature_offline_token(client.clone(), ip.to_string())),
        )
    }
    pub fn signature_password_token(
        client: states::Client,
        ip: String,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        warp::path("password_token")
            .and(warp::post())
            .and(warp::body::content_length_limit(1024 * 16))
            .and(with_grpc(client))
            .and(with_ip(ip))
            .and(warp::body::json())
            .and_then(|client, ip, token_create: PasswordTokenCreate| async {
                grpc_to_http(grpc_create_password_token(client, ip, token_create).await).await
            })
    }
    pub fn signature_refresh_token(
        client: states::Client,
        ip: String,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        warp::path("refresh_token")
            .and(warp::post())
            .and(with_grpc(client))
            .and(with_ip(ip))
            .and(warp::body::json())
            .and_then(|client, ip, token_params: RefreshTokenParams| async {
                let token_create = RefreshTokenCreate {
                    ip,
                    access_token: token_params.access_token,
                    refresh_token: token_params.refresh_token,
                };
                grpc_to_http(grpc_create_refresh_token(client, token_create).await).await
            })
    }
    pub fn signature_offline_token(
        client: states::Client,
        ip: String,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        warp::path("offline_token")
            .and(warp::post())
            .and(with_grpc(client))
            .and(with_ip(ip))
            .and(warp::body::json())
            .and_then(|client, ip, token: TrustTokenCreate| async {
                grpc_to_http(grpc_create_trust_token(client, ip, token).await).await
            })
    }
    pub fn push_routes(
        client: states::Client,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        let staging_routes = warp::path("push_staging")
            .and(push_staging_avro_bulk(client.clone()).or(push_staging_json_bulk(client.clone())));
        let online_routes = warp::path("push_online")
            .and(push_online_avro_bulk(client.clone()).or(push_online_json_bulk(client.clone())));
        staging_routes.or(online_routes)
    }

    pub fn push_staging_avro_bulk(
        client: states::Client,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        with_grpc(client)
            .and(warp::header::<String>("Authorization"))
            .and(warp::path!(
                "bucket" / String / "path" / String / "avro_bulk"
            ))
            .and(warp::post())
            .and(warp::body::content_length_limit(1024 * 16))
            .and(warp::body::bytes())
            .and_then(
                |client, header_authorization, bucket, path, body_bytes: bytes::Bytes| {
                    let create = AvroBulkCreate {
                        header_authorization,
                        compression: "zstd".into(),
                        bucket,
                        path,
                        body: body_bytes.to_vec(),
                    };
                    async {
                        grpc_to_http(grpc_create_staging_avro_bulk(client, create).await).await
                    }
                },
            )
    }

    pub fn push_staging_json_bulk(
        client: states::Client,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        with_grpc(client)
            .and(warp::header::<String>("Authorization"))
            .and(warp::path!(
                "bucket" / String / "path" / String / "json_bulk"
            ))
            .and(warp::post())
            .and(warp::body::content_length_limit(1024 * 1000))
            .and(warp::body::bytes())
            .and_then(
                |client, header_authorization, bucket, path, body_bytes: bytes::Bytes| {
                    let create = JsonBulkCreate {
                        header_authorization,
                        compression: "none".into(),
                        bucket,
                        path,
                        body: body_bytes.to_vec(),
                    };
                    async {
                        grpc_to_http(grpc_create_staging_json_bulk(client, create).await).await
                    }
                },
            )
    }

    pub fn push_online_avro_bulk(
        client: states::Client,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        with_grpc(client)
            .and(warp::header::<String>("Authorization"))
            .and(warp::path!(
                "bucket" / String / "path" / String / "avro_bulk"
            ))
            .and(warp::post())
            .and(warp::body::content_length_limit(1024 * 16))
            .and(warp::body::bytes())
            .and_then(
                |client, header_authorization, bucket, path, body_bytes: bytes::Bytes| {
                    let create = AvroBulkCreate {
                        header_authorization,
                        compression: "zstd".into(),
                        bucket,
                        path,
                        body: body_bytes.to_vec(),
                    };
                    async { grpc_to_http(grpc_create_online_avro_bulk(client, create).await).await }
                },
            )
    }
    pub fn push_online_json_bulk(
        client: states::Client,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        with_grpc(client)
            .and(warp::header::<String>("Authorization"))
            .and(warp::path!(
                "bucket" / String / "path" / String / "json_bulk"
            ))
            .and(warp::post())
            .and(warp::body::content_length_limit(1024 * 16))
            .and(warp::body::bytes())
            .and_then(
                |client, header_authorization, bucket, path, body_bytes: bytes::Bytes| {
                    let create = JsonBulkCreate {
                        header_authorization,
                        compression: "none".into(),
                        bucket,
                        path,
                        body: body_bytes.to_vec(),
                    };
                    async { grpc_to_http(grpc_create_online_json_bulk(client, create).await).await }
                },
            )
    }

    pub fn query_routes(
        client: states::Client,
        ip: &str,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        warp::path("query").and(
            warp::path("health")
                .map(|| StatusCode::OK)
                .or(query_avsc_schema(client.clone(), ip)),
        )
    }
    pub fn query_avsc_schema(
        client: states::Client,
        ip: &str,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        with_grpc(client)
            .and(with_ip(ip.to_string()))
            // .and(warp::header::<String>("Authorization"))
            .and(warp::path!("bucket" / String / "path" / String / "meta"))
            .and_then(|client, ip: String, bucket: String, path: String| {
                let read = SchemaRead {
                    schema: format!("avsc.{}.{}", bucket, path.strip_suffix(".avsc").unwrap()),
                };
                async { grpc_to_http_json(grpc_read_raw_schema(client, ip, read).await).await }
            })
    }
    pub fn meta_routes(
        client: states::Client,
        ip: &str,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        warp::path("meta").and(
            warp::path("health")
                .map(|| StatusCode::OK)
                .or(read_raw_schema(client.clone(), ip))
                .or(read_avsc_schema(client.clone(), ip))
                .or(read_avsc_schema_simple(client.clone(), ip)),
        )
    }
    pub fn read_raw_schema(
        client: states::Client,
        ip: &str,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        with_grpc(client)
            .and(with_ip(ip.to_string()))
            .and(warp::path!("raw_schema" / String))
            .and_then(|client, ip: String, schema_id: String| {
                let read = SchemaRead { schema: schema_id };
                async { grpc_to_http_json(grpc_read_raw_schema(client, ip, read).await).await }
            })
    }

    pub fn read_avsc_schema(
        client: states::Client,
        ip: &str,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        with_grpc(client)
            .and(with_ip(ip.to_string()))
            .and(warp::path!("avsc_schema" / String))
            .and_then(|client, ip, schema_id: String| {
                let read = SchemaRead { schema: schema_id };
                async { grpc_to_http_json(grpc_read_avsc_schema(client, ip, read).await).await }
            })
    }
    pub fn read_avsc_schema_simple(
        client: states::Client,
        ip: &str,
    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
        with_grpc(client)
            .and(with_ip(ip.to_string()))
            .and(warp::path!("avsc_schema_simple" / String))
            .and_then(|client, ip, schema_id: String| {
                let read = SchemaRead { schema: schema_id };
                async { grpc_to_message_json(grpc_read_avsc_schema(client, ip, read).await).await }
            })
    }
}

mod handlers {
    use super::*;
    #[derive(super::Deserialize, super::Serialize)]
    pub struct JsonStatus {
        code: u8,
        message: String,
        details: Option<String>,
    }
    pub fn grpc_code(code: Code) -> u8 {
        use Code::*;
        match code {
            Ok => 0,
            Cancelled => 1,
            Unknown => 2,
            InvalidArgument => 3,
            DeadlineExceeded => 4,
            NotFound => 5,
            AlreadyExists => 6,
            PermissionDenied => 7,
            ResourceExhausted => 8,
            FailedPrecondition => 9,
            Aborted => 10,
            OutOfRange => 11,
            Unimplemented => 12,
            Internal => 13,
            Unavailable => 14,
            DataLoss => 15,
            Unauthenticated => 16,
            _ => 99,
        }
    }

    impl From<super::Status> for JsonStatus {
        fn from(status: super::Status) -> Self {
            JsonStatus {
                code: grpc_code(status.code()),
                message: status.message().into(),
                details: None,
            }
        }
    }

    pub async fn grpc_to_http<T>(
        grpc_result: Result<Response<T>, Status>,
    ) -> Result<impl super::Reply, super::Infallible>
    where
        T: super::Serialize,
    {
        match grpc_result {
            Ok(resp) => Ok(json(resp.get_ref())),
            Err(status) => {
                let status = JsonStatus::from(status);
                Ok(json(&status))
            }
        }
    }

    pub async fn grpc_to_http_json(
        grpc_result: Result<Response<DefaultReply>, Status>,
    ) -> Result<impl super::Reply, super::Infallible> {
        match grpc_result {
            Ok(resp) => {
                let reply = resp.get_ref();
                let json_reply = json!({
                    "code": reply.code ,
                    "message": serde_json::from_str(&reply.message).ok().unwrap_or(json!({"raw":&reply.message})),
                    "details": reply.details,
                    "origin": reply.origin
                });
                // Ok(unimplemented!())
                Ok(json(&json_reply))
            }
            Err(status) => {
                let status = JsonStatus::from(status);
                Ok(json(&status))
            }
        }
    }
    pub async fn grpc_to_message_json(
        grpc_result: Result<Response<DefaultReply>, Status>,
    ) -> Result<impl super::Reply, super::Infallible> {
        match grpc_result {
            Ok(resp) => {
                let reply = resp.get_ref();
                let json_reply = serde_json::from_str(&reply.message)
                    .ok()
                    .unwrap_or(json!({"raw":&reply.message}));
                // Ok(unimplemented!())
                Ok(json(&json_reply))
            }
            Err(status) => {
                let status = JsonStatus::from(status);
                Ok(json(&status))
            }
        }
    }

    pub async fn grpc_create_password_token(
        client: states::Client,
        ip: String,
        token_create: PasswordTokenCreate,
    ) -> Result<Response<SignatureTokenReply>, Status> {
        verify_lan_ip(ip)?;
        let mut clone_client = client.clone();

        clone_client
            .signature
            .create_password_token(PasswordTokenCreate {
                app_secret: "chinapex".to_string(),
                ..token_create
            })
            .await
    }
    pub async fn grpc_create_refresh_token(
        client: states::Client,
        token_create: RefreshTokenCreate,
    ) -> Result<Response<SignatureTokenReply>, Status> {
        let mut clone_client = client.clone();
        clone_client
            .signature
            .create_refresh_token(token_create)
            .await
    }

    pub async fn grpc_create_trust_token(
        client: states::Client,
        ip: String,
        token_create: TrustTokenCreate,
    ) -> Result<Response<SignatureTokenReply>, Status> {
        verify_lan_ip(ip)?;
        let mut clone_client = client.clone();
        clone_client
            .signature
            .create_trust_token(token_create)
            .await
    }

    pub async fn grpc_create_staging_avro_bulk(
        client: states::Client,
        bulk_create: AvroBulkCreate,
    ) -> Result<Response<DefaultReply>, Status> {
        //let bs_str = String::from_utf8_lossy(&bulk_create.body) ;
        let bs_str = base64::encode(&bulk_create.body);
        let avro_gate = AvroGate {
            // schema: "avdl.data_buffer:dcSdkPlatformBulk".to_string(),
            schema: "avsc.dcSdkPlatformBulk".to_string(),
            body: bs_str.to_string(),
        };

        let mut clone_client = client.clone();
        let meta_reply = clone_client
            .meta
            .gate_bin_avro(avro_gate)
            .await?
            .into_inner();
        if meta_reply.code != 0 {
            Ok(Response::new(meta_reply))
        } else {
            clone_client
                .push
                .create_staging_avro_bulk(bulk_create)
                .await
        }
    }
    pub async fn grpc_create_staging_json_bulk(
        client: states::Client,
        bulk_create: JsonBulkCreate,
    ) -> Result<Response<DefaultReply>, Status> {
        let bs_str = String::from_utf8_lossy(&bulk_create.body);
        let avro_gate = AvroGate {
            // schema: "avdl.data_buffer:dcSdkPlatformBulk".to_string(),
            schema: "avsc.dcSdkPlatformBulk".to_string(),
            body: bs_str.to_string(),
        };
        let mut clone_client = client.clone();
        let meta_reply = clone_client
            .meta
            .gate_json_avro(avro_gate)
            .await?
            .into_inner();
        if meta_reply.code != 0 {
            Ok(Response::new(meta_reply))
        } else {
            clone_client
                .push
                .create_staging_json_bulk(bulk_create)
                .await
        }
    }

    pub async fn grpc_create_online_avro_bulk(
        client: states::Client,
        bulk_create: AvroBulkCreate,
    ) -> Result<Response<DefaultReply>, Status> {
        let mut clone_client = client.clone();
        clone_client.push.create_online_avro_bulk(bulk_create).await
    }

    pub async fn grpc_create_online_json_bulk(
        client: states::Client,
        bulk_create: JsonBulkCreate,
    ) -> Result<Response<DefaultReply>, Status> {
        let mut clone_client = client.clone();
        clone_client.push.create_online_json_bulk(bulk_create).await
    }

    pub async fn grpc_read_raw_schema(
        client: states::Client,
        ip: String,
        read: SchemaRead,
    ) -> Result<Response<DefaultReply>, Status> {
        verify_lan_ip(ip)?;
        let mut clone_client = client.clone();
        clone_client.meta.read_raw_schema(read).await
    }
    pub async fn grpc_read_avsc_schema(
        client: states::Client,
        ip: String,
        read: SchemaRead,
    ) -> Result<Response<DefaultReply>, Status> {
        verify_lan_ip(ip)?;
        let mut clone_client = client.clone();
        clone_client.meta.read_avsc_schema(read).await
    }
}

mod gossip {}

mod states {
    use super::*;
    use std::iter::Iterator;

    #[derive(Debug, Clone)]
    pub struct Client {
        pub push: PushClient<Channel>,
        pub signature: SignatureClient<Channel>,
        pub meta: MetaClient<Channel>,
        pub query: QueryClient<Channel>,
    }
    pub async fn mk_client<'a>(
        push_hosts: &'a str,
        signature_hosts: &'a str,
        meta_hosts: &'a str,
        query_hosts: &'a str,
    ) -> Client {
        let push_endpoints = push_hosts
            .split(',')
            .map(|a| Channel::from_shared(a.to_owned()).unwrap());

        let signature_endpoints = signature_hosts
            .split(',')
            .map(|a| Channel::from_shared(a.to_owned()).unwrap());

        let meta_endpoints = meta_hosts
            .split(',')
            .map(|a| Channel::from_shared(a.to_owned()).unwrap());

        let query_endpoints = query_hosts
            .split(',')
            .map(|a| Channel::from_shared(a.to_owned()).unwrap());

        Client {
            push: PushClient::new(Channel::balance_list(push_endpoints)),
            signature: SignatureClient::new(Channel::balance_list(signature_endpoints)),
            meta: MetaClient::new(Channel::balance_list(meta_endpoints)),
            query: QueryClient::new(Channel::balance_list(query_endpoints)),
        }
    }
}

mod utils {
    use super::*;
    pub fn resolve_socket_addr(nodes: &str) -> Option<IpAddr> {
        let ips = get_if_addrs::get_if_addrs()
            .unwrap()
            .iter()
            .map(|interface| interface.ip())
            .collect::<HashSet<_>>();
        let ip = nodes
            .split(',')
            .filter_map(|x| x.parse::<IpAddr>().ok())
            .find(|x| ips.contains(&x));
        ip
    }
    pub fn build_nodes(nodes: &str, port: &str) -> String {
        itertools::join(
            nodes.split(',').map(|x| format!("http://{}:{}", x, port)),
            ",",
        )
    }
    pub fn verify_lan_ip(ip: String) -> Result<(), Status> {
        let my_ip = ip.parse::<Ipv4Addr>().unwrap();
        if !my_ip.is_private() {
            return Err(Status::unauthenticated("global ip forbidden!"));
        }
        Ok(())
    }
}

enum EitherBody<A, B> {
    Left(A),
    Right(B),
}

impl<A, B> http_body::Body for EitherBody<A, B>
where
    A: http_body::Body + Send + Unpin,
    B: http_body::Body<Data = A::Data> + Send + Unpin,
    A::Error: Into<Error>,
    B::Error: Into<Error>,
{
    type Data = A::Data;
    type Error = Box<dyn std::error::Error + Send + Sync + 'static>;

    fn is_end_stream(&self) -> bool {
        match self {
            EitherBody::Left(b) => b.is_end_stream(),
            EitherBody::Right(b) => b.is_end_stream(),
        }
    }

    fn poll_data(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
        match self.get_mut() {
            EitherBody::Left(b) => Pin::new(b).poll_data(cx).map(map_option_err),
            EitherBody::Right(b) => Pin::new(b).poll_data(cx).map(map_option_err),
        }
    }

    fn poll_trailers(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
        match self.get_mut() {
            EitherBody::Left(b) => Pin::new(b).poll_trailers(cx).map_err(Into::into),
            EitherBody::Right(b) => Pin::new(b).poll_trailers(cx).map_err(Into::into),
        }
    }
}

fn map_option_err<T, U: Into<Error>>(err: Option<Result<T, U>>) -> Option<Result<T, Error>> {
    err.map(|e| e.map_err(Into::into))
}
